package com.atguigu.flink.chapter3;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2020/12/19 9:59
 */
public class Flink01_chapter3_TaskChain {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.disableOperatorChaining();
        env
          .socketTextStream("hadoop162", 9999)
          .flatMap(new FlatMapFunction<String, String>() {
              @Override
              public void flatMap(String value, Collector<String> out) throws Exception {
                  String[] split = value.split(" ");
                  for (String word : split) {

                      out.collect(word);
                  }
              }
          })
          .map(new MapFunction<String, Tuple2<String, Long>>() {
              @Override
              public Tuple2<String, Long> map(String value) throws Exception {
                  return Tuple2.of(value, 1L);
              }
          })
          .keyBy(new KeySelector<Tuple2<String, Long>, String>() {
              @Override
              public String getKey(Tuple2<String, Long> value) throws Exception {
                  return value.f0;
              }
          })
          .sum(1)
          .print("sum");

        env.execute();
    }
}
/*
整体控制:
    在env上进行控制
算子单独控制:
    段子的后面设置并行度


算子>环境整体>参数>配置文件

任务链:
    .startNewChain()  和前面的切开  算子上

    .disableChaining()  和前后都切开 算子上

    env.disableOperatorChaining();  在整个job禁用任务链
 */